package day2

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object PairRDD {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\新建文件夹\\HCIA初级\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))

    //PairRDD：键值对RDD，在Spark中，只能是一个二元组才能转换为PairRDD
    //*ByKey、*ByValue => 都是属于PairRDD独有的，*By => 表示为共有
    val pair = sc.parallelize(List((1,1),(1,2),(1,3),(1,4),(1,5),(2,1),(2,2),(2,3),(2,4),(2,5)))
    val pair_int = sc.parallelize(List(1,2,3,4,5)).map(i => (i,1))
    val pair_string = sc.parallelize(List("hello world","hello scala","hello spark")).map(i => {
      val list = i.split(" ")
      (list(0),i)
    })

    //    val test = sc.parallelize(List("hello world hello scala hello spark")).map(i => {
    //      val list = i.split(" ")
    //      (list(1),(list(3),list(5)))
    //    })
    //      .foreach(println)

    //转换算子
    //mapValues()算子：处理PairRDD当中的value
    val pair_mapValues = pair.mapValues(println(_))
    //使用map()算子实现pair_mapValues方式
    val pair_map = pair.map{case (k: Int,v) => println(v)}

    //reduceByKey()：接受一个函数，对于具有相同键的值进行操作
    //reduceByKey()算子返回一个PairRDD，输出的是一个相同键的值合并完整之后的值(k1,vs1),(k2,vs2)
    val pair_StringInt = sc.parallelize(
      List(('a',1),('b',2),('c',3),('d',4),('a',5),('b',1),('c',2),('d',3),('a',4),('b',5))
    )
    val pair_StringInt_reduceByKey = pair_StringInt.reduceByKey(_+_)

    //countByValue()算子：将一个标准的RDD，相同的值当作key，value输出为key的个数
    //将标准的RDD转换成为PairRDD，具体示例查看wordCount.scala

    //combineByKey()算子：操作方法与aggregate()类似,求同一个key的value操作
    /*
     * 将相同key的values放置在同一个分区
     * 实现分区内操作和分区间操作
     * combineByKey(a)(b,c)
     * a：读取到的value内容，并转化为Pair
     * b：分区内操作
     * c：分区间操作
     */
    //(List((1,1),(1,2),(1,3),(1,4),(1,5),(2,1),(2,2),(2,3),(2,4),(2,5))
    pair.combineByKey(
      v => (v,1),
      (k: (Int,Int),v: Int) => (k._1 + v , k._2 + 1),
      (k: (Int,Int),v: (Int,Int)) => (k._1+v._1,v._2+k._2)
    )
    //求相同key当中平均值
    //      .mapValues(i => println(i._1/i._2)).count()

    //sortByKey()算子：强制要求按照key进行排序；默认为true，表示为升序
    //    pair_StringInt.sortByKey(ascending= false).foreach(println)
    //sortBy()算子：自定义排序操作
    //    pair_StringInt.sortBy(i => i._1,ascending = false).foreach(println)

    //groupByKey()算子：相同的key进行合并，值进行合并 => (a,1) (a,2) -groupByKey->(a,Seq(1,2))
    //    pair_StringInt.groupByKey().foreach(println)
    //Seq类型：集合类型

    //实现RDD合并，通过将相同的key进行value合并
    //join()全连接，将所有的key提取出来，然后相同key的值合并
    val pair_join = sc.parallelize(
      List(('a',1),('b',2),('c',3),('d',4),('e',5),('b',1),('c',2),('d',3),('e',4),('b',5))
    )
    //('a',1),('b',2),('c',3),('d',4),('a',5),('b',1),('c',2),('d',3),('a',4),('b',5)
    //    pair_StringInt.join(pair_join).foreach(println)
    //leftOuterJoin()左连接
    //    pair_StringInt.leftOuterJoin(pair_join).foreach(println)
    //rightOuterJoin()右连接
    pair_StringInt.rightOuterJoin(pair_join).foreach(println)
    //Option类型：None,Some

  }
}